package com.shujia.streaming

import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Code01WordCount {
  def main(args: Array[String]): Unit = {

    // 如果要使用SparkStreaming 需要获取其操作对象 => StreamingContext
    // 如果创建StreamingContext对象那么需要SparkContext对象
    // local[2] 表示本地模式 并且设置并发数为2 此项为必须项
    val sc = new SparkContext(new SparkConf().setMaster("local[2]").setAppName("wordCountStreaming"))

    // Durations.seconds(5) 表示每5秒启动一个计算任务
    val streamingContext: StreamingContext = new StreamingContext(sc, Durations.seconds(5))
    // 可以监听socket端口获取数据 master:8888 在Linux中可以通过nc -lk master 8888 启动
    val stream: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 8888)

    // 获取输入流对象 可以对数据进行做计算操作
    stream
      .flatMap {
        case words => words.split(" ")
      }
      .map {
        case word => (word, 1)
      }
      .reduceByKey(
        (v1: Int, v2: Int) => (v1 + v2)
      ).print()


    // 需要启动 streamingContext
    streamingContext.start()
    // 等待停止信号 <- 手动停止
    streamingContext.awaitTermination()

    // Streaming程序是 7 * 24 小时不间断运行，那么不需停止
    //    streamingContext.stop()

  }
}
